package realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import realtime.bean.TrafficPageViewBean;
import realtime.util.ClickHouseUtil;
import realtime.util.DateFormatUtil;
import realtime.util.MyKafkaUtil;
import realtime.util.MyWindowFunction;

import java.time.Duration;

public class Dws_TrafficPageView {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        /*env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointTimeout(10000);
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/flinkCDC");
        env.setStateBackend(new HashMapStateBackend());*/

        System.setProperty("HADOOP_USER_NAME", "root");

        DataStreamSource<String> kafkaStream = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer("dwd_traffic_page_log", "traffic_page_view"));

        SingleOutputStreamOperator<JSONObject> jsonStream = kafkaStream.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String value, Collector<JSONObject> out) throws Exception {
                if (value != null) {
                    JSONObject jsonObject = JSON.parseObject(value);
                    out.collect(jsonObject);

                }
            }
        });

        SingleOutputStreamOperator<JSONObject> jsonStreamWithWatermark = jsonStream.assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(10))
                .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                    @Override
                    public long extractTimestamp(JSONObject element, long recordTimestamp) {
                        return element.getLong("ts");
                    }
                }));

        KeyedStream<JSONObject, String> keyedStream = jsonStreamWithWatermark.keyBy(key -> key.getJSONObject("common").getString("mid"));

        SingleOutputStreamOperator<TrafficPageViewBean> javaBeanStream = keyedStream.flatMap(new RichFlatMapFunction<JSONObject, TrafficPageViewBean>() {
            private ValueState<String> valueState;

            @Override
            public void open(Configuration parameters) throws Exception {
                StateTtlConfig build = StateTtlConfig.newBuilder(Time.days(1))
                        .setUpdateType(StateTtlConfig.UpdateType.OnReadAndWrite)
                        .build();

                ValueStateDescriptor<String> valueStateDescriptor = new ValueStateDescriptor<>("valueState", String.class);
                valueStateDescriptor.enableTimeToLive(build);
                valueState = getRuntimeContext().getState(valueStateDescriptor);

            }

            @Override
            public void flatMap(JSONObject value, Collector<TrafficPageViewBean> out) throws Exception {
                String oldDate = valueState.value();
                Long ts = value.getLong("ts");
                String todayDate = DateFormatUtil.toDate(ts);

                long uv = 0;
                if (oldDate == null || oldDate.equals(todayDate)) {
                    uv = 1;
                }

                String lastPageId = value.getJSONObject("page").getString("last_page_id");
                String itemType = value.getJSONObject("page").getString("item_type");
                String pageId = value.getJSONObject("page").getString("page_id");
                long sv = 0;
                long bs = 0;
                if (lastPageId == null && pageId != null) {
                    sv = 1;
                    if (itemType == null) {
                        bs = 1;
                    }
                }

                out.collect(TrafficPageViewBean.builder()
                        .sc(value.getJSONObject("common").getString("sc"))
                        .svCt(sv)
                        .pvCt(1L)
                        .durSum(value.getJSONObject("page").getLong("during_time"))
                        .bounceSum(bs)
                        .isNew(value.getJSONObject("common").getString("is_new"))
                        .uvCt(uv)
                        .ch(value.getJSONObject("common").getString("ch"))
                        .build());
            }
        });

        SingleOutputStreamOperator<TrafficPageViewBean> reduce = javaBeanStream.windowAll(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10)))
                .reduce(new ReduceFunction<TrafficPageViewBean>() {
                    @Override
                    public TrafficPageViewBean reduce(TrafficPageViewBean value1, TrafficPageViewBean value2) throws Exception {

                        value1.setUvCt(value1.getUvCt() + value2.getUvCt());
                        value1.setSvCt(value1.getSvCt() + value2.getSvCt());
                        value1.setDurSum(value1.getDurSum() + value2.getDurSum());
                        value1.setPvCt(value1.getPvCt() + value2.getPvCt());
                        value1.setBounceSum(value1.getBounceSum() + value2.getBounceSum());

                        return value1;
                    }
                }, new AllWindowFunction<TrafficPageViewBean, TrafficPageViewBean, TimeWindow>() {
                    @Override
                    public void apply(TimeWindow window, Iterable<TrafficPageViewBean> values, Collector<TrafficPageViewBean> out) throws Exception {
                        TrafficPageViewBean next = values.iterator().next();
                        MyWindowFunction.getWindowFunction(window, next);
                        out.collect(next);
                    }
                });


        reduce.print(">>");
        reduce.addSink(ClickHouseUtil.getSinkFunction("insert into dws_traffic_sc_page_view_window values(?,?,?,?,?,?,?,?,?,?,?)"));
        env.execute("Dws_TrafficPageView");

    }
}
